package mr

import (
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"sync"
	"time"
)

// CoordinatorStatus 定义了整批作业的不同状态
type CoordinatorStatus string

const (
	Idle     CoordinatorStatus = "协调器空闲"             // 空闲状态
	Mapping  CoordinatorStatus = "协调器在 Mapping 阶段"  // 映射阶段
	Reducing CoordinatorStatus = "协调器在 Reducing 阶段" // 归约阶段
	Finished CoordinatorStatus = "协调器已完成"           // 作业已完成
)

// Coordinator 协调器
type Coordinator struct {
	// 作业状态
	Status  CoordinatorStatus // "idle", "mapping", "reducing", "completed"
	nReduce int
	// Map任务列表及其状态
	mapTasks []Task // 包含所有Map任务的信息

	// Reduce任务列表及其状态
	reduceTasks []Task // 包含所有Reduce任务的信息

	mu sync.Mutex // 用于同步访问共享资源：比如访问结构体内的字段

}

type WorkerInfo struct {
	workerId    string    // Worker的地址或ID
	lastSeen    time.Time // 上次心跳时间
	TasksActive []*Task   // 当前正在执行的任务列表
}

// Your code here -- RPC handlers for the worker to call.

// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {
	err := rpc.Register(c)
	if err != nil {
		return
	}
	rpc.HandleHTTP()
	l, e := net.Listen("tcp", ":"+coordinatorPort)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

// main/mrcoordinator.go calls Done() periodically to find out
// if the entire Task has finished.
func (c *Coordinator) Done() bool {

	// Your code here.

	return c.allReduceTasksDone()
}

// allMapTasksDone 判断map任务是否全部完成
func (c *Coordinator) allMapTasksDone() bool {
	var rest []int
	for _, t := range c.mapTasks {
		if t.Status != Completed {
			rest = append(rest, t.TaskID)
		}
	}
	if len(rest) == 0 {
		return true
	} else {
		fmt.Printf("还有%v个 map 任务未完成,分别为%v\n", len(rest), rest)
		return false
	}
}

func (c *Coordinator) allReduceTasksDone() bool {
	var rest []int
	for _, t := range c.reduceTasks {
		if t.Status != Completed {
			rest = append(rest, t.TaskID)
		}
	}
	if len(rest) == 0 {
		return true
	} else {
		fmt.Printf("还有%v个 Reduce 任务未完成,分别为%v\n", len(rest), rest)
		return false
	}
}

// PrintStatus 打印当前状态
func (c *Coordinator) PrintStatus() {
	fmt.Printf("Coordinator 状态:")

	switch c.Status {
	case Idle:
		fmt.Println("Job尚未开始")
	case Mapping:
		totalMapTasks := len(c.mapTasks)
		fmt.Printf("Job正处于 Mapping 状态，共 %v 个Task\n", totalMapTasks)
		for i, t := range c.mapTasks {
			fmt.Printf("      Task %v: 状态：%v\n", i, t.Status)
		}
	case Reducing:
		totalReduceTasks := len(c.reduceTasks)
		fmt.Printf("Job正处于 Reducing 状态，共 %v 个Task\n", totalReduceTasks)
		for i, t := range c.reduceTasks {
			fmt.Printf("      Task %v: 状态：%v\n", i, t.Status)
		}
	case Finished:
		fmt.Println("Job已经完成")
	default:
		fmt.Println("未知状态")
	}
}

//// create a Coordinator.
//// main/mrcoordinator.go calls this function.
//// nReduce is the number of reduce tasks to use.
//func MakeCoordinator(files []string, nReduce int) *Coordinator {
//	c := Coordinator{}
//
//	// Your code here.
//
//	c.server()
//	return &c
//}

// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce Tasks to use.
func MakeCoordinator(files []string, nReduce int) *Coordinator {

	c := Coordinator{
		Status: Idle,
		//todo 合理设置Task数量
		mapTasks:    make([]Task, len(files)),
		reduceTasks: make([]Task, nReduce),
		//workers:              make(map[string]*WorkerInfo),
		//inputFiles:           files,
		//outputDir:            "output",
		//maxMapTasks:          MaxTaskNum,
		//maxReduceTasks:       nReduce,
		nReduce: nReduce,
	}
	if len(files) != 0 {
		c.Status = Mapping
	}

	// 初始化Map Tasks
	for i, file := range files {
		c.mapTasks[i] = Task{TaskType: MapTask, FileName: file, TaskID: i, Status: Pending, ReduceNum: nReduce}
	}
	for i := 0; i < nReduce; i++ {
		c.reduceTasks[i] = Task{TaskType: ReduceTask, TaskID: i, Status: Pending, ReduceNum: nReduce}
	}

	c.server()
	return &c
}

func (c *Coordinator) MonitorTimeouts() {
	for {
		time.Sleep(time.Second)
		c.mu.Lock()
		for i := range c.mapTasks {
			if c.mapTasks[i].Status == Active && time.Since(c.mapTasks[i].StartTime) > TaskTimeOut {
				c.mapTasks[i].Status = Pending
			}
		}
		for i := range c.reduceTasks {
			if c.reduceTasks[i].Status == Active && time.Since(c.reduceTasks[i].StartTime) > TaskTimeOut {
				c.reduceTasks[i].Status = Pending
			}
		}
		c.mu.Unlock()
	}
}

// ********************** RPC方法 ********************************

func (c *Coordinator) GetTask(req GetTaskArgs, res *GetTaskReply) error {
	c.mu.Lock()

	defer c.mu.Unlock()

	var task Task
	// 根据当前作业状态选择合适的任务列表

	if c.Status == Mapping {
		for taskId, t := range c.mapTasks {
			if t.Status == Pending {
				task = t
				task.Status = Active
				c.mapTasks[taskId].StartTime = time.Now()
				c.mapTasks[taskId].Status = Active // 更新原始任务列表中的状态
				c.PrintStatus()
				break
			}
		}
	} else if c.Status == Reducing {
		for i, t := range c.reduceTasks {
			if t.Status == Pending {
				task = t
				task.Status = Active
				c.reduceTasks[i].StartTime = time.Now() // 更新任务列表中的开始执行时间
				c.reduceTasks[i].Status = Active        // 更新原任务列表中的状态
				break
			}
		}
	}
	res.ReduceCount = c.nReduce
	// 将找到的任务设置到响应中
	res.Task = task
	return nil
}

func (c *Coordinator) CompleteTask(req CompleteTaskArgs, res *CompleteTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	fmt.Println("CompleteTask调用，参数为：", req)
	if req.TaskType == MapTask {
		c.mapTasks[req.TaskId].Status = Completed
		ok := c.allMapTasksDone()
		if ok {
			c.Status = Reducing
		}
	} else if req.TaskType == ReduceTask {
		c.reduceTasks[req.TaskId].Status = Completed
		ok := c.allReduceTasksDone()
		if ok {
			c.Status = Finished
		}
	}
	res.Err = nil
	return nil
}
